
#include "stdafx.h"

#include "FASTServerApplication.h"

#include <cstdlib>
#include <iostream>

// boost
#include <boost/bind.hpp>
#include <boost/smart_ptr.hpp>
#include <boost/asio.hpp>
#include <boost/thread/thread.hpp>

// QuickFAST
#include <Codecs\DataDestination.h>
#include <Codecs\FieldInstruction.h>
#include <Codecs\Encoder.h>


#include <Messages\FieldAscii.h>
#include <Messages\FieldSet.h>
#include <Messages\Message.h>

using namespace QuickFAST;

using boost::asio::ip::tcp;

const int max_read_length = 1024;
const int max_write_length = 1024 * 128;

///////////////////////
// ServerApplication

FASTServerApplication::FASTServerApplication()
{
}

FASTServerApplication::~FASTServerApplication()
{
}

bool FASTServerApplication::init(int argc, char * argv[])
{
	bool baseResult = FASTApplication::init(argc, argv);

	if (!baseResult)
		return false;

	pEncoder_ = Codecs::EncoderPtr(new Codecs::Encoder(GetRegistry()));

	return true;
}

void FASTServerApplication::AddField(Messages::FieldSet& message,
                                  const char* fieldName,
                                  const std::string& fieldValue)
{
  Messages::FieldIdentityCPtr 
    fieldNamePtr(new QuickFAST::Messages::FieldIdentity(fieldName));
  Messages::FieldCPtr fieldValuePtr = Messages::FieldAscii::create(fieldValue);

  message.addField(fieldNamePtr,fieldValuePtr);
}


void FASTServerApplication::EncodeTestMessage(Codecs::DataDestination& destination, unsigned int blockNumber, unsigned int messagesNumber)
{
	static unsigned int seqNumber = 0;
	seqNumber++;
	Messages::FieldSet message(12);

	AddField(message, "BeginString", "FIXT.1.1");
	AddField(message, "MsgType", "A");
	AddField(message, "Field1", boost::lexical_cast<std::string>(blockNumber).c_str());
	AddField(message, "Field2", boost::lexical_cast<std::string>(messagesNumber).c_str());
	AddField(message, "Field3", boost::lexical_cast<std::string>(seqNumber).c_str());
	AddField(message, "Field4", "4444");
	AddField(message, "Field5", "5555");
	AddField(message, "Field6", "6666");
	AddField(message, "Field7", "7777");
	AddField(message, "Field8", "8888");
	AddField(message, "Field9", "9999");

	pEncoder_->encodeMessage(destination, 125, message);

}

void FASTServerApplication::SendTestFASTMessages(socket_ptr sock, unsigned int blockNumbers, unsigned int messagesPerBlock)
{
	Codecs::DataDestination destination;

	for(unsigned int block = 0; block < blockNumbers; ++ block)
	{
		Codecs::DataDestination::BufferHandle blockSizeBuffer = destination.startBuffer();
		for(unsigned int message = 0; message < messagesPerBlock; ++ message)
		{
			EncodeTestMessage(destination, block, message);
		}

		// calculate message size and put it to the buffer at the beginning of the message

		// iterate all message's buffers
		size_t messageSize = 0;
		for (size_t i = blockSizeBuffer + 1; i < destination.size(); ++i)
		{
			messageSize += destination[i].size();
		}

		// encode block size
		WorkingBuffer tempBuffer;
		destination.selectBuffer(blockSizeBuffer);
		Codecs::FieldInstruction::encodeUnsignedInteger(destination, tempBuffer, messageSize);
	}

	WorkingBuffer wholeBlockWorkingBuffer;
	destination.toWorkingBuffer(wholeBlockWorkingBuffer);

	size_t wholeSize = wholeBlockWorkingBuffer.size();
	std::cout << "SendTestFASTMessages: sending " << wholeSize << std::endl;

	boost::asio::write(*sock, boost::asio::buffer(wholeBlockWorkingBuffer.begin(), wholeSize));
}

void FASTServerApplication::SendDataForMode(socket_ptr sock)
{
	if(GetMode() == "1")
	{
		std::cout << "SendDataForMode: mode 1 - Testing StreamingAssembler with more then one message per block" << std::endl;
		SendTestFASTMessages(sock, 1 /* block numbrs*/, 3 /* messages per block*/);
		SendTestFASTMessages(sock, 1 /* block numbrs*/, 3 /* messages per block*/);
	}
	else if (GetMode() == "2")
	{
		std::cout << "SendDataForMode: mode 2 - Testing MessagePerPacketAssembler with long-long block sending via TCP" << std::endl;
		SendTestFASTMessages(sock, 10 /* block numbrs*/, 100 /* messages per block*/);
	}
	else if (GetMode() == "3")
	{
		std::cout << "SendDataForMode: mode 3 - Testing SpecialAssembler with long-long block sending via TCP" << std::endl;
		SendTestFASTMessages(sock, 100 /* block numbrs*/, 1 /* messages per block*/);
		SendTestFASTMessages(sock, 100 /* block numbrs*/, 2 /* messages per block*/);
		SendTestFASTMessages(sock, 100 /* block numbrs*/, 3 /* messages per block*/);
		SendTestFASTMessages(sock, 100 /* block numbrs*/, 10 /* messages per block*/);
		SendTestFASTMessages(sock, 100 /* block numbrs*/, 100 /* messages per block*/);
	}
	else
		std::cerr << "SendDataForMode: Incorrect mode:" << GetMode() << std::endl;
}

void FASTServerApplication::Session(socket_ptr sock)
{
  try
  {
    for (;;)
    {
      char read_data[max_read_length];

      boost::system::error_code error;
      size_t length = sock->read_some(boost::asio::buffer(read_data), error);
      if (error == boost::asio::error::eof)
        break; // Connection closed cleanly by peer.
      else if (error)
        throw boost::system::system_error(error); // Some other error.

	  SendDataForMode(sock);
//      boost::asio::write(*sock, boost::asio::buffer(data, length));
    }
  }
  catch (std::exception& e)
  {
    std::cerr << "Exception in thread: " << e.what() << "\n";
  }
}

void FASTServerApplication::Server(boost::asio::io_service& io_service, short port)
{
  tcp::acceptor a(io_service, tcp::endpoint(tcp::v4(), port));
  for (;;)
  {
    socket_ptr sock(new tcp::socket(io_service));
    a.accept(*sock);
    boost::thread t(boost::bind(&FASTServerApplication::Session, this, sock));
  }
}

int FASTServerApplication::run()
{
	int result = 0;

	try
	{
		boost::asio::io_service io_service;
		Server(io_service, std::atoi(GetPort().c_str()));
	}
	catch (std::exception & e)
	{
		std::cerr << "FASTServerApplication::run(): Exception:" << e.what() << std::endl;
		result = -1;
	}

	return result;
}

